package com.vcg.community.filter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.ClasspathPropertiesFileCredentialsProvider;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreamsClient;
import com.amazonaws.services.dynamodbv2.model.*;
import com.amazonaws.services.dynamodbv2.streamsadapter.AmazonDynamoDBStreamsAdapterClient;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
import com.fivepx.utils.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 *1. adapterRun 通过适配器调用kcl程序获取kinesis流信息(生产环境下用此种方式)
 * 2.apiRun 通过低级API来获取kinesis流信息(测试环境用此种方式)
 * @author felayman
 * @since 16/1/6
 */
@Component
public class DynamoWorker {

    private static Logger logger = Logger.getLogger(DynamoWorker.class);

    @Autowired
    private AmazonDynamoDBClient dynamoDBClient;
    @Autowired
    private AmazonDynamoDBStreamsClient amazonDynamoDBStreamsClient;

        private  Worker worker;
        private  KinesisClientLibConfiguration workerConfig;
        private  IRecordProcessorFactory recordProcessorFactory;
        private  AmazonDynamoDBStreamsAdapterClient adapterClient;
        private  AWSCredentialsProvider credentialsProvider;
        private  AmazonCloudWatchClient cloudWatchClient;
        private  String serviceName = "dynamodb";
        private  final String dynamodbEndpoint = "dynamodb.ap-northeast-1.amazonaws.com";
        private  final String streamsEndpoint = "streams.dynamodb.ap-northeast-1.amazonaws.com";
        private  final String tableName = "vcg-activity-slave-test";
        private final String applicationName = "kcl-lease-manager";

        public int  adapterRun() throws InterruptedException {
            credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();
            recordProcessorFactory = new StreamsRecordProcessorFactory();
            adapterClient = new AmazonDynamoDBStreamsAdapterClient(credentialsProvider, new ClientConfiguration());
            adapterClient.setEndpoint(streamsEndpoint);
            dynamoDBClient = new AmazonDynamoDBClient(credentialsProvider, new ClientConfiguration());
            dynamoDBClient.setEndpoint(dynamodbEndpoint);
            cloudWatchClient = new AmazonCloudWatchClient(credentialsProvider, new ClientConfiguration());
            DescribeTableResult describeTableResult = dynamoDBClient.describeTable(tableName);
            String streamArn = describeTableResult.getTable().getLatestStreamArn();
            String workdId = UUID.randomUUID().toString().replace("-","");
            workerConfig = new KinesisClientLibConfiguration(applicationName,streamArn, credentialsProvider, workdId)
                    .withMaxRecords(1000)
                    .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);
            worker = new Worker(recordProcessorFactory, workerConfig, adapterClient, dynamoDBClient, cloudWatchClient);
            Thread t = new Thread(worker);
            t.start();
            t.join();
            return 0;
        }

        public int apiRun(){
            DescribeTableResult describeTableResult = dynamoDBClient.describeTable(tableName);
            TableDescription tableDescription = describeTableResult.getTable();
            String myStreamArn = tableDescription.getLatestStreamArn();
            StreamSpecification myStreamSpec = tableDescription.getStreamSpecification();
            DescribeStreamResult describeStreamResult = amazonDynamoDBStreamsClient.describeStream(new DescribeStreamRequest().withStreamArn(myStreamArn));
            StreamDescription streamDescription = describeStreamResult.getStreamDescription();
            String streamArn = streamDescription.getStreamArn();
            List<Shard> shards = streamDescription.getShards();
            for (Shard shard : shards) {
                String shardId = shard.getShardId();
                logger.info("Processing " + shardId + " from stream " + streamArn);
                GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest()
                        .withStreamArn(myStreamArn)
                        .withShardId(shardId)
                        .withShardIteratorType(ShardIteratorType.TRIM_HORIZON);
                GetShardIteratorResult getShardIteratorResult = amazonDynamoDBStreamsClient.getShardIterator(getShardIteratorRequest);
                String nextItr = getShardIteratorResult.getShardIterator();
                while (nextItr != null) {
                    GetRecordsResult getRecordsResult = amazonDynamoDBStreamsClient.getRecords(new GetRecordsRequest().withShardIterator(nextItr));
                    List<Record> records = getRecordsResult.getRecords();
                    for (Record record : records) {
                        StreamRecord streamRecord = record.getDynamodb();
                        String eventName = record.getEventName();
                        //请在这里进行断点调试,这里会接受来自于于dynamodb的数据
                        /**
                         * 1.接受来自于于dynamodb的数据
                         * 2.数据经过DataSwitcher进行格式转换(内部转换流程有些复杂)
                         * 3.转换成vcg所需要的格式后会通过kafka转发给我vcg系统
                         */
                        DataSwitcher dataSwitcher = new DataSwitcher();
                        if (eventName.equals("REMOVE")){
                            Map<String, AttributeValue>  oldMap =  streamRecord.getOldImage();
                            Map <String,Object> dataMap =  MapUtils.toSingleMap(oldMap);
                            JSONObject result = JSON.parseObject(JSON.toJSONString(dataMap));
                            dataSwitcher.transform(eventName,result.toJSONString());
                        }else {
                        Map<String, AttributeValue> stringAttributeValueMap=  streamRecord.getNewImage();
                        Map <String,Object> dataMap =  MapUtils.toSingleMap(stringAttributeValueMap);
                        JSONObject result = JSON.parseObject(JSON.toJSONString(dataMap));
                        dataSwitcher.transform(eventName,result.toJSONString());
                    }
                }
                    nextItr = getRecordsResult.getNextShardIterator();
                    if (StringUtils.isNotBlank(nextItr)){
                    }
                }
            }
            return 0;
        }

}
